package com.sinosoft.app.utils.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sinosoft.app.common.kafka.vo.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 解析消息
 */
@Component
public final class ParsingMsg2PostgresqlUtils {

    public static final Logger log = LoggerFactory.getLogger(ParsingMsg2PostgresqlUtils.class);

    public static final String DateTime = "DateTime";
    public static final String DATE = "Date";//yyyy-MM-dd
    public static final String NULL = "null";
    private static final String year = "year";
    private static final String month = "month";
    private static final String day = "day";
    private static final String hour = "hour";
    private static final String minute = "minute";
    private static final String second = "second";
    private static ObjectMapper mapper = new ObjectMapper();

    /**
     * 事件类型
     */
    public static final String UPDATE = "u";
    public static final String INSERT = "c";
    public static final String DELETE = "d";
    public static final String FULL = "r";
    /**
     * name
     */
    private static final String DATE_TYPE = "io.debezium.time.Date"; // int32
    private static final String TIME_TYPE = "io.debezium.time.Time"; // int32
    private static final String MICROTIME_TYPE = "io.debezium.time.MicroTime"; // int 64
    private static final String TIMESTAMP_TYPE = "io.debezium.time.Timestamp"; // int 64
    private static final String MICROTIMETIMESTAMP_TYPE = "io.debezium.time.MicroTimestamp"; // int 64
    private static final String DATE_CONNECT_TYPE = "org.apache.kafka.connect.data.Date";
    private static final String TIME_CONNECT_TYPE = "org.apache.kafka.connect.data.Time";
    private static final String TIMESTAMP_CONNECT_TYPE = "org.apache.kafka.connect.data.Timestamp";

    private static DataTypeWithName dtwithname = new DataTypeWithName();
    private static DatatypeWithoutName dtwithoutname = new DatatypeWithoutName();

    /**
     * 数据类型，没有标注指定名字
     */
    public static class DatatypeWithoutName {
        public static final int BOOLEAN = 0; // true false
        public static final int INT = 1; // int16-64 ,float,double
        public static final int STRING = 2;
        public static final int BYTES = 3;
        public static final int ARRAY = 4;
        public static final int MAP = 5;
        public static final int STRUCT = 6;
        public static final int FLOAT = 7;
        public static final int DOUBLE = 8;

        public static final int TIMESTAMP6 = 17;
        public static final int INTERVAL = 14;
        public static final int TIMESTAMP15 = 15;
        public static final int TIMESTAMP16 = 16;
        public static final int DATE = 100;
        public static final int TIME = -1;

        private int dataType(String type) {
            switch (type) {
                case "boolean":
                    return BOOLEAN;
                case "int16":
                    return INT;
                case "int32":
                    return INT;
                case "int64":
                    return INT;
                case "float32":
                    return FLOAT;
                case "float64":
                    return DOUBLE;
                case "string":
                    return STRING;
                case "bytes":
                    return BYTES;
                case "array":
                    return ARRAY;
                case "map":
                    return MAP;
                case "struct":
                    return STRUCT;
            }
            return -1; // 错误类型
        }
    }

    /**
     * 数据类型指定名字 例如time,date
     */
    static final class DataTypeWithName {
        /**
         * 当time.precision.mode配置属性设置为adaptive
         */
        static final int ADA_DATE = 9; // int32
        static final int ADA_TIME = 10; // int32
        static final int ADA_MICROTIME = 11; // int64
        static final int ADA_TIMESTAMP = 12; // int64
        static final int ADA_MIRCROTIMESTAMP = 13; // int64
        /**
         * 当time.precision.mode配置属性设置为 adaptive_time_microseconds
         */
        static final int ATM_DATE = 14; // int32
        static final int ATM_TIME = 15; // int64
        static final int ATM_TIMESTAMP = 16; // int64
        static final int ATM_MIRCROTIMESTAMP = 17; // int64
        /**
         * 当time.precision.mode配置属性设置为 connect
         */
        static final int CON_DATE = 18; // int32
        static final int CON_TIME = 19; // int64
        static final int CON_TIMESTAMP = 20; // int64

        public   static final int TIMESTAMP6 = 17;
        public   static final int INTERVAL = 14;
        public  static final int TIMESTAMP15 = 15;
        public  static final int DATE = 100;
        public static final int TIMESTAMPZ = -1;
        /**
         * other
         */
        static final int JSON = 12;

        private int dataType(String name, String config) {
            if (StringUtils.equals(config, "adaptive")) { // 默认设置
                switch (name) {
                    case DATE_TYPE:
                        return ADA_DATE;
                    case TIME_TYPE:
                        return ADA_TIME;
                    case TIMESTAMP_TYPE:
                        return ADA_TIMESTAMP;
                    case MICROTIME_TYPE:
                        return ADA_MICROTIME;
                    case MICROTIMETIMESTAMP_TYPE:
                        return ADA_MIRCROTIMESTAMP;
                }
            }
            if (StringUtils.equals(config, "adaptive_time_microseconds")) {
                switch (name) {
                    case DATE_TYPE:
                        return ATM_DATE;
                    case TIMESTAMP_TYPE:
                        return ATM_TIMESTAMP;
                    case MICROTIME_TYPE:
                        return ATM_TIME;
                    case MICROTIMETIMESTAMP_TYPE:
                        return ATM_MIRCROTIMESTAMP;
                }
            }
            if (StringUtils.equals(config, "connect")) {
                switch (name) {
                    case DATE_CONNECT_TYPE:
                        return CON_DATE;
                    case TIMESTAMP_CONNECT_TYPE:
                        return CON_TIMESTAMP;
                    case TIME_CONNECT_TYPE:
                        return CON_TIME;
                }
            }
            return -1; // 不支持的类型
        }
    }

    public NewEvent parseShemaMsg(String key, String value) { // 未作异常处理，后续处理
//        JSONObject keyjson = JSONObject.parseObject(key);
//        JSONObject valuejson = JSONObject.parseObject(value);
//        Event event = new Event(); // 封装成事件对象返回
//        parseKey(keyjson, event);
//        parseValue(valuejson, event);

        NewEvent event = new NewEvent();
        event.setKey(ParsingMsg2PostgresqlUtils.getMap(key));
        event.setValue(ParsingMsg2PostgresqlUtils.getMap(value));
        return event;
    }

    public Event parseShemaMsg(String value) { // 未作异常处理，后续处理
        JSONObject valuejson = JSONObject.parseObject(value);
        Event event = new Event(); // 封装成事件对象返回
        parseValue(valuejson, event);
        return event;
    }

    public <T> T Parse2Object(Event event, Class<T> clazz) throws IllegalAccessException, InstantiationException, InvocationTargetException {
        Object obj = clazz.newInstance();
        Method[] methods = obj.getClass().getMethods();
        for (ValueField field : event.getValue()) {
            for (Method m : methods) {
                if (("set" + field.getField()).toLowerCase().equals(m.getName().toLowerCase())) {
                    log.info("Method :" + m.getName());
                    log.info("Method field:" + JSON.toJSONString(field));
                    log.info("Method value:" + field.getValue());
                    if (field.getValue() != null) {
                        switch (field.getTypeclass()) {
                            case DatatypeWithoutName.TIMESTAMP6:
                                String interval = DateUtil.parseInterval((long) field.getValue());
                                m.invoke(obj, interval);
                                break;
                            case DatatypeWithoutName.INTERVAL:
                                LocalDate localDate = DateUtil.parseDate((Long.parseLong(field.getValue().toString())));
                                m.invoke(obj, localDate.toString());
                                break;
                            case DatatypeWithoutName.TIMESTAMP15:
                                LocalTime localTime = DateUtil.parseTime(Long.parseLong(field.getValue().toString()));
                                m.invoke(obj, localTime.toString());
                                break;
                            case -1:
                                m.invoke(obj, field.getValue().toString());
                                break;
                            default:
                                m.invoke(obj, field.getValue());
                                break;
                        }
                    }
                }
            }
            log.info("T :" + JSON.toJSONString(obj));
        }
        return clazz.cast(obj);
    }

    /**
     * 适配主键
     *
     * @param key
     */
    private Event parseKey(JSONObject key, Event event) {
        SchemaKey schemakey = key.getObject("schema", SchemaKey.class);
        Map<String, Object> payloadkey = key.getJSONObject("payload").getInnerMap();
        List<ValueField> keyfields = new ArrayList<>(); // 优化
        schemakey.getFields().stream().forEach(field -> {
            ValueField valuefield = new ValueField();
            valuefield.setField(field.getField());
            valuefield.setOptional(field.isOptional());
            valuefield.setType(field.getType());
            valuefield.setName(field.getName());
            valuefield.setValue(payloadkey.get(field.getField()));
            valuefield.setTypeclass(parseDataType(field.getName(), field.getType())); // 设置类型
            keyfields.add(valuefield);
        });
        event.setKey(keyfields);
        return event;
    }

    /**
     * value值
     *
     * @param value
     * @param event
     */
    private Event parseValue(JSONObject value, Event event) {
        SchemaValue schemavalue = value.getObject("schema", SchemaValue.class);
        JSONObject payloadvalue = value.getJSONObject("payload");
        String op = (String) payloadvalue.get("op");
        long time = (long) payloadvalue.get("ts_ms");
        event.setOp(op);
        event.setTime(time);
        Map<String, Object> source = payloadvalue.getJSONObject("source").getInnerMap(); // 表名
        event.setTable((String) source.get("table"));
        if (StringUtils.equals(INSERT, op)) { // 创建
            Map<String, Object> map = payloadvalue.getJSONObject("after").getInnerMap();
            return parseValueByop(map, schemavalue, event, "after");
        }
        if (StringUtils.equals(UPDATE, op)) { // 修改
            Map<String, Object> map = payloadvalue.getJSONObject("after").getInnerMap();
            return parseValueByop(map, schemavalue, event, "after");
        }
        if (StringUtils.equals(DELETE, op)) { // 删除
            Map<String, Object> map = payloadvalue.getJSONObject("before").getInnerMap();
            return parseValueByop(map, schemavalue, event, "before");
        }
        if (StringUtils.equals(FULL, op)) { // 快照读取
            Map<String, Object> map = payloadvalue.getJSONObject("after").getInnerMap();
            return parseValueByop(map, schemavalue, event, "after");
        }
        return event;
    }

    private Event parseValueByop(Map<String, Object> map, SchemaValue value, Event event, String filed) {
        List<ValueField> valuefields = new ArrayList<>();
        value.getFields().stream().filter((structField -> StringUtils // 筛选有效字段
                .equals(structField.getField(), filed))).forEach((structField -> {
            structField.getFields().stream().forEach(field -> {
                ValueField valuefield = new ValueField();
                valuefield.setType(field.getType());
                valuefield.setOptional(field.isOptional());
                valuefield.setField(field.getField());
                valuefield.setName(field.getName());
                valuefield.setTypeclass(parseDataType(field.getName(), field.getType()));
                valuefield.setValue(map.get(field.getField()));
                valuefields.add(valuefield);
            });
        }));
        event.setValue(valuefields);
        return event;
    }

    private int parseDataType(String name, String type) {
        if (StringUtils.isBlank(name)) {
            return dtwithoutname.dataType(type);
        } else {
            return dtwithname.dataType(name, "adaptive_time_microseconds"); // 需要改成配置文件形式
        }
    }

    public static List<Field> getMap(String value) {
        List<Field> rMap = new ArrayList<>();
        if (StringUtils.isBlank(value)){
            return rMap;
        }
        Map<String, Map<String, Object>> map;
        try {
            map = mapper.readValue(value, Map.class);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("解析value出错! ", e);
        }
        map.forEach((k, v) -> {
            if (v == null || NULL.equals(v)) {
                return;
            }
            Map<String, Object> mapV = (v);
            String subK = mapV.keySet().stream().findFirst().get();
            Object subV = mapV.get(subK);

            if (DateTime.equals(subK)) {
                subV = parseDateTime((Map<String, Object>) subV);
            } else if (DATE.equals(subK)) {
                // add by wangqian 兼容Date类型转换 2018-11-12
                subV = parseDate((Map<String, Object>) subV);
            }
            rMap.add(new Field(k, String.valueOf(subV)));
        });
        return rMap;
    }

    private static String parseDateTime(Map<String, Object> map) {
        LocalDateTime datetime = LocalDateTime.of(string2Int(year, map), string2Int(month, map) , string2Int(day, map),
                string2Int(hour, map), string2Int(minute, map), string2Int(second, map));
        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        return datetime.format(format);
    }

    /**
     * Date类型转换
     *
     * @param map
     * @return
     * @author Wangqian 20181112
     */
    private static String parseDate(Map<String, Object> map) {
        LocalDate datetime = LocalDate.of(string2Int(year, map), string2Int(month, map) , string2Int(day, map));
        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd");
        return datetime.format(format);
    }

    private static int string2Int(String name, Map<String, Object> map) {
        return Integer.parseInt(map.get(name).toString());
    }

}
